package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo04DSL {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName(this.getClass.getSimpleName.replace("$", ""))
      // 控制在SparkSQL中进行Shuffle操作时默认的分区数，默认值为200，相当于会启动200个Task进行处理
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    /**
     * DSL常见的操作：select、where（having）、groupBy、orderBy、limit、union、join、常见的函数
     */


    val stuDF: DataFrame = spark.read.format("csv").option("sep", ",")
      .schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/stu/students.txt")

    // select：将需要的列查询出来，或者是使用函数对列进行处理
    stuDF.select("id", "name")
    //      .show()
    stuDF.select($"id", $"name")
    //      .show()
    // Scala中的if是关键字，所以SparkSQL Functions没有包含if方法，可以用when方法替代
    stuDF.select($"id", $"name", when($"gender" === "男", 1).otherwise(0) as "new_gender")
    //      .show()


    // where：过滤
    stuDF.where("clazz like '理科%'")
    //      .show()

    stuDF.where($"clazz" like "理科%")
    //      .show()

    stuDF.where($"age" === 22)
    //      .show()

    stuDF.where($"age" =!= 22)
    //      .show()

    // groupBy 分组，一般后续会跟聚合操作
    // 统计班级人数
    stuDF
      .groupBy($"clazz")
      .agg(countDistinct($"id") as "cnt", max("age") as "max_age", avg("age") as "avg_age")
      .where($"cnt" >= 100) // 相当于SQL中的having，代码执行顺序没有嵌套只有先后
    //      .show() // 最终结果只有groupBy的字段以及聚合结果会保留，其它没有使用的列会去掉


    // orderBy：用于排序
    // limit：限制返回的条数
    stuDF
      .groupBy($"clazz")
      .agg(countDistinct($"id") as "cnt")
      .orderBy($"cnt".desc)
      .limit(3)
    //      .show()

    // withColumnRenamed：对列进行重命名
    // withColumn：给数据增加一列
    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,subject_id String,score Int")
      .load("spark/data/stu/score.txt")
    // 统计班级总分top3
    scoreDF
      .groupBy($"id")
      .agg(sum($"score") as "sumScore")
      .withColumnRenamed("sumScore", "sum_score")
      .join(stuDF, "id")
      //      .select(
      //        $"id"
      //        , $"name"
      //        , $"age"
      //        , $"clazz"
      //        , $"sum_score"
      //        , row_number() over Window.partitionBy("clazz").orderBy($"sum_score".desc) as "rn"
      //      )
      // 给DF数据增加一列，适合搭配窗口函数的使用
      .withColumn("rn", row_number() over Window.partitionBy($"clazz").orderBy($"sum_score".desc))
      .where($"rn" <= 3)
    //      .show()

    // union：合并两个DF，作用和SQL中的union all相同
    val stuDF01: DataFrame = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)
    val stuDF02: DataFrame = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)
    stuDF01.union(stuDF02)
    //      .show(50) // 默认不会去重
    stuDF01.union(stuDF02).distinct() // 可以再接上distinct去重
    //      .show(50)


    // 在SparkSQL中关联时：
    // 如果两个DF的关联字段名字一样，则只能通过字符串方式指定关联字段
    // 如果两个DF的字段名字不一样，则只能通过列表达式的方式指定关联字段
    // inner join：内连接
    stuDF.join(scoreDF, "id") // 默认内连接
    //      .show()

    // 当关联字段同名时无法直接使用列表达式，会出现Reference 'id' is ambiguous错误
    //    stuDF.join(scoreDF, $"id").show()
    val renameStuDF: DataFrame = stuDF.withColumnRenamed("id", "stuId")
    renameStuDF.join(scoreDF, $"stuId" === $"id")
    //      .show()

    renameStuDF.join(scoreDF, $"id" === $"stuId", "inner")
    //      .show()

    stuDF.join(scoreDF, List("id"), "inner")
    //      .show()

    // 多条件关联时：关联名一样时通过List传递多个关联条件，不同时通过列表达式and列表达式......传递
    stuDF.join(scoreDF, List("id"), "left")
    //      .show()
    stuDF.join(scoreDF, List("id"), "right")
    //      .show()
    renameStuDF.join(scoreDF, $"stuId" === $"id", "right")
    //      .show()
    renameStuDF.join(scoreDF, $"stuId" === $"id", "full")
    //      .show()

    // join：inner、outer（left、right、full）


    //    while(true){
    //
    //    }

  }

}
